package com.atguigu.day08;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;

// 实时对账
public class Example1_2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.getConfig().setAutoWatermarkInterval(1L);

        SingleOutputStreamOperator<OrderEvent> stream1 = env
                .addSource(new SourceFunction<OrderEvent>() {
                    @Override
                    public void run(SourceContext<OrderEvent> ctx) throws Exception {
                        OrderEvent e1 = new OrderEvent("order-1", "app-zhifu", 1000L);
                        Thread.sleep(200L);

                        ctx.collect(e1);
                        OrderEvent e2 = new OrderEvent("order-2", "app-zhifu", 2000L);
                        Thread.sleep(200L);
                        ctx.collect(e2);
                    }

                    @Override
                    public void cancel() {

                    }
                }).assignTimestampsAndWatermarks(WatermarkStrategy.<OrderEvent>forMonotonousTimestamps().withTimestampAssigner((SerializableTimestampAssigner<OrderEvent>) (element, recordTimestamp) -> element.timestamp));

        SingleOutputStreamOperator<OrderEvent> stream2 = env
                .addSource(new SourceFunction<OrderEvent>() {
                    @Override
                    public void run(SourceContext<OrderEvent> ctx) throws Exception {
                        OrderEvent e1 = new OrderEvent("order-1", "weixin-zhifu", 4000L);
                        ctx.collect(e1);
                        Thread.sleep(200L);
                        OrderEvent e2 = new OrderEvent("order-3", "weixin-zhifu", 6000L);
                        Thread.sleep(200L);

                        ctx.collect(e2);
                        OrderEvent e3 = new OrderEvent("order-2", "weixin-zhifu", 8000L);
                        Thread.sleep(200L);
                        ctx.collect(e3);
                    }

                    @Override
                    public void cancel() {

                    }
                }).assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<OrderEvent>forMonotonousTimestamps()
                                .withTimestampAssigner((SerializableTimestampAssigner<OrderEvent>) (element, recordTimestamp) -> element.timestamp));

        stream1.keyBy(r -> r.orderId)
                .connect(stream2.keyBy(r -> r.orderId))
                .process(new MatchFunction())
                .print();

        env.execute();
    }

    public static class MatchFunction extends CoProcessFunction<OrderEvent, OrderEvent, String> {
        // 如果app支付信息先到达，那么保存在appEvent中，并等待微信支付的到达5秒钟
        private ValueState<OrderEvent> appEvent;
        // 如果weixin支付信息先到达，那么保存在weixinEvent中，并等待app支付的到达5秒钟
        private ValueState<OrderEvent> weixinEvent;
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            appEvent = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("app", Types.POJO(OrderEvent.class)));
            weixinEvent = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("weixin", Types.POJO(OrderEvent.class)));
        }

        @Override
        public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
            System.out.println(ctx.timerService().currentWatermark());
            if (weixinEvent.value() != null) {
                // weixin支付先到达
                out.collect(value.orderId + "对账成功");
                weixinEvent.clear();
            } else {
                // app支付来了，对应的weixin支付没来
                appEvent.update(value);
                ctx.timerService().registerEventTimeTimer(value.timestamp + 5000L);
            }
        }

        @Override
        public void processElement2(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
            System.out.println(ctx.timerService().currentWatermark());
            if (appEvent.value() != null) {
                out.collect(value.orderId + "对账成功");
                appEvent.clear();
            } else {
                weixinEvent.update(value);
                ctx.timerService().registerEventTimeTimer(value.timestamp + 5000L);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            if (appEvent.value() != null) {
                out.collect(appEvent.value().orderId + "对应的微信支付没来，对账失败");
                appEvent.clear();
            }
            if (weixinEvent.value() != null) {
                out.collect(weixinEvent.value().orderId + "对应的app支付没来，对账失败");
                weixinEvent.clear();
            }
        }
    }

    public static class OrderEvent {
        public String orderId;
        public String type;
        public Long timestamp;

        public OrderEvent() {
        }

        public OrderEvent(String orderId, String type, Long timestamp) {
            this.orderId = orderId;
            this.type = type;
            this.timestamp = timestamp;
        }

        @Override
        public String toString() {
            return "OrderEvent{" +
                    "orderId='" + orderId + '\'' +
                    ", type='" + type + '\'' +
                    ", timestamp=" + timestamp +
                    '}';
        }
    }
}
